package com.my.service.task.map;

import com.my.service.task.entity.SexPredictionInfo;
import com.my.service.task.logistic.Logistic;
import com.my.service.task.util.HbaseUtils;
import org.apache.flink.api.common.functions.MapFunction;

import java.util.ArrayList;
import java.util.Random;

public class SexPredictSaverMap implements MapFunction<String, SexPredictionInfo> {
    private ArrayList<Double> weights = null;

    public SexPredictSaverMap(ArrayList<Double> weights) {
        this.weights = weights;
    }

    @Override
    public SexPredictionInfo map(String s) throws Exception {
        String[] temps = s.split("\t");
        int userid = Integer.parseInt(temps[0]);
        long ordernum = Long.parseLong(temps[1]);//订单的总数
        long orderfre = Long.parseLong(temps[4]);//隔多少天下单
        int manclothes = Integer.parseInt(temps[5]);//浏览男装次数
        int womenclothes = Integer.parseInt(temps[6]);//浏览女装的次数
        int childclothes = Integer.parseInt(temps[7]);//浏览小孩衣服的次数
        int oldmanclothes = Integer.parseInt(temps[8]);//浏览老人的衣服的次数
        double avramount = Double.parseDouble(temps[9]);//订单平均金额
        int producttimes = Integer.parseInt(temps[10]);//每天浏览商品数
        ArrayList<String> as = new ArrayList<>();
        as.add(ordernum + "");
        as.add(orderfre + "");
        as.add(manclothes + "");
        as.add(womenclothes + "");
        as.add(childclothes + "");
        as.add(oldmanclothes + "");
        as.add(avramount + "");
        as.add(producttimes + "");
        String sexflag = Logistic.classifyVector(as, weights);
        String sexstring = sexflag == "0" ? "女" : "男";
        String tablename = "userflaginfo";
        String rowkey = userid + "";
        String famliyname = "baseinfo";
        String colum = "sex"; //预测出的性别
        HbaseUtils.putdata(tablename, rowkey, famliyname, colum, sexstring);
        return null;
    }
}

